//
//  sync_channel.cpp
//  Amqp_test
//
//  Created by HNXF on 2022/7/28.
//

#include "sync_channel.h"
#include <iostream>
#include <thread>
sync_channel::sync_channel(boost::asio::io_service &io_service, AMQP::Connection &connect)
    : io_service_(io_service)
    , channel_(&connect)
    , reliable_(channel_)
    , operation_finished_(false)
    , is_error_state_(false)
{
    
}

sync_channel::~sync_channel() {
    stop();
}

void sync_channel::stop()
{
    on_error("Operation aborted.");
}

void sync_channel::set_envelope_format(const std::string& event_type_name, AMQP::Envelope& envelope)
    {
        envelope.setContentType("application/json");
        envelope.setPersistent(true);

        AMQP::Table header_table;
        header_table.set("proto", event_type_name.c_str());
        envelope.setHeaders(header_table);
    }

void sync_channel::publish(const std::string &topic, const std::string &partition_key, const std::string &event_type_name, const std::string &message) {
//    std::cout <<  ": sync_channel::publish begin" << std::endl;
    std::cout << std::this_thread::get_id() << ": sync_channel::publish begin\n";
    
    std::unique_lock publish_lock(publish_mutex_);
    std::unique_lock operation_lock(operation_mutex_);
    operation_finished_ = false;
    
    if (!is_error_state_) {
        std::cout << std::this_thread::get_id() << ": sync_channel::publish locked\n";

        io_service_.post([this, topic, partition_key, event_type_name, message]()
        {
            AMQP::Envelope envelope(message.data(), message.size());
            set_envelope_format(event_type_name, envelope);

            reliable_.publish(topic, partition_key, envelope)
                .onAck([this]() {
                std::cout << std::this_thread::get_id() << ": sync_channel::publish onAck\n";
                on_success();
            }).onLost([this]() {
                std::cout << std::this_thread::get_id() << ": sync_channel::publish onLost\n";
                on_error("Message failed to publish!");
            });

            //channel_.confirmSelect()
        //    .onSuccess([this, &topic, &partition_key, &message]()
        //{
        //    std::cout << "sync_channel::publish onSuccess\n";
        //    channel_.publish(topic, partition_key, message);
        //})
        //    .onAck([this](uint64_t /*delivery_tag*/, bool /*multiple*/)
        //{
        //    std::cout << "sync_channel::publish onAck\n";
        //    onSuccess();
        //})
        //    .onNack([this](uint64_t /*delivery_tag*/, bool /*multiple*/, bool /*requeue*/)
        //{
        //    std::cout << "sync_channel::publish onNAck\n";
        //    onError("Message failed to publish!");
        //});
        
        });
    }
    wait_operation_finish(std::move(operation_lock));
}

void sync_channel::wait_operation_finish(std::unique_lock<std::recursive_mutex> lock)
{
    std::cout << std::this_thread::get_id() << ": sync_channel::wait_operation_finish begin\n";
    operation_finished_cv_.wait(lock, [this]()
    {
        return operation_finished_ || is_error_state_;
    });
    std::cout << std::this_thread::get_id() << ": sync_channel::wait_operation_finish finished\n";
    if (is_error_state_)
    {
        std::string message("sync_channel error: ");
        message.append(error_message_);
        std::cout << std::this_thread::get_id() << ": sync_channel::wait_operation_finish error: " << message << "\n";
        throw std::runtime_error(message);
    }
}

void sync_channel::on_success() {
    std::unique_lock lock(operation_mutex_);
    operation_finished_ = true;
    operation_finished_cv_.notify_all();
}

void sync_channel::on_error(const std::string& message)
{
    std::unique_lock lock(operation_mutex_);
    operation_finished_ = true;
    is_error_state_ = true;
    error_message_ = message;
    operation_finished_cv_.notify_all();
}
